/*-
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 2016, 2017 Oracle and/or its affiliates.  All rights reserved.
 *
 * ex_slice - Use the Berkeley DB feature slices in a benchmarking application.
 *
 *				Introduction
 *
 *	This is an example benchmark for the Berkeley DB feature "slices".
 *	It is a multithreaded, potentially multiprocess program which performs
 *	random DB->get(), DB->put() and DB()->del() calls over a generated
 *	dataset. It has many options for measuring times and adjusting BDB
 *	parameters in order to find the best configuration settings for a
 *	particular dataset size, number of threads, etc.
 *
 *	The slices feature is not automatically included; it must be enabled
 *	when configuring BDB:
 *
 *		cd build_unix
 *		../dist/configure --enable-slices ...
 *		make all ex_slice
 *	To run it with 20 slices and 6 threads, removing the SLICE20 directory:
 *		ex_slice -R -h SLICE20 -s 20  -w6	
 *
 *	There are many options (see print_help()),  available by specifying an
 *	invalid option, or just "ex_slice -".
 *
 *				The Workload
 *
 *	A number of threads are started, which each repeatedly perform either
 *	a read or some kind of write operation until a total operation count is
 *	reached, or until a specified number of seconds have passed, or until
 *	the program is interrupted by SIGINT.
 *
 *	Each operation selects a key at random. A reader thread does a DB->get()
 *	with that key, silently ignoring whether or not it is present. A writer
 *	thread first probes the sliced database for that key. If it is missing
 *	then a key-value pair is inserted; otherwise the existing data is
 *	deleted or (with the -u 'do updates' option) the value portion is
 *	changed. Thus by default the database will eventually grow to be half full.
 *
 *	By default there are no reader threads and 4 writer threads, 16 slices,
 *	doing 1,000,000 writer transactions in a database of 1,000,000 records.
 *	It displays throughput statistics every 10 seconds and at the end of the
 *	run.
 *
 *	The keys are obtained from "`../test/tcl/wordlist", or "wordlist". Since
 *	the requested number of distinct keys ( -n <int> ) is usually larger
 *	than the number of words in this file (~10K), the additional keys are
 *	generated by appending a record number to the word, e.g., "avow-821688".
 *
 *				Options
 *
 *	Per-slice resource usage:
 *	    File Descriptors OPEN_MAX (sh: ulimit -n; csh: limit descriptors)
 *		Each slice requires at least 4 file descriptors, plus one for
 *		each open database. This test uses ony one database, so a
 *		typical large server will permit enough open file descriptors
 *		for hundreds of slices. If there are EMFILE errors, try
 *		increasing ``ulimit -n'' before runnign this program..
 *
 *	    Memory
 *		Each slice is given a Berkeley DB cache size of 200 bytes per
 *		record, to minimize the effect of any disk I/O system limitations.
 *		It can be overwritten by adding -c <config-line> options to the
 *		command line. To give each slice a single 1GB cache:
 *			ex_slice .... -c "slice all set_cachesize 1 0 1"
 *		
 *
 * program name: ex_slice
 * environment: SLICE
 * database: slice.db
 *
 * $Id$
 */

#include <sys/types.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <sys/resource.h>
#include <sys/wait.h>
#include <assert.h>

#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#ifdef _WIN32
extern int getopt(int, char * const *, const char *);
#else
#include <unistd.h>
#endif

#include <db.h>

#define MEGABYTE	(1024 * 1024)
#define GIGABYTE	(MEGABYTE * 1024ULL)

/* Maximum number of argv parameters to be passed to -M# sub-processes. */
#define MAXARGC		40

/*
 * Use this to suppress the "unused argument" warnings generated by some
 * compilers.
 */
#define COMPQUIET(v, n) { v = (n); v = (v); }
#define ABS(v)		((v) < 0 ? (-v) : (v))

extern int sched_yield __P((void));		/* Pthread yield function. */

int	db_init __P((const char *, DB_ENV **, u_int32_t));
void   *deadlock __P((void *));
void	fatal __P((const char *, int, ...));
void	onint __P((int));
int	main __P((int, char *[]));
int	open_database __P((DB_ENV *, const char *, DB **, u_int32_t));
int	preload __P((DB *, int));
int	reader __P((int));
void	stats __P((long, int));
void	slice_stats __P((void));
void   *trickle __P((void *));
void   *tstart __P((void *));
int	usage __P((int, const char *));
void	word __P((void));
int	writer __P((int));
int	verify __P((void));

typedef enum SliceKey {
	SliceKeyWhole,
	SliceKeySingle,
	SliceKeyMultiple
} SliceKeyType;

/*
 * These variables contain the command line options.
 */
char	 *ConfigLines = NULL;	/* -c Lines to write to initial DB_CONFIG. */
int	  Checkpoint = 0;	/* -C Cause a checkpoint at end of the run. */
int	  Duration;		/* -d <#seconds> Set the end time of the run. */
int	  DbType = DB_BTREE;	/* -H Changes Database type to hash */
char	 *Home = "SLICE";	/* -h <home directory for the environment> */
SliceKeyType KeyType =		/* -k <s|m> Change slice key from the default */
	  SliceKeyWhole;	/* to the 1st or 1st&3rd '-'  fields. */
char	 *LogWriteMode = "write_nosync"; /* -L DB_TXN_{NO,WRITE_NO,}SYNC */
int	  LockerCache = 0;	/* -L set_thread_count() for locker cache */
int	  MultiProc = 0;	/* -m <#procs> Run several processes. */
long long NRecords = 1000000;	/* -n # of distinct records. */
int	  NoCreate = 0;		/* -N Do not create the environment; use when
				      running several instances in parallel. */
int	  OverflowSize = 0;	/* -O <extra size of large, overflow records. */
long long OpCount;		/* -oN Do this many operations. */
int	  Pagesize = 8192;	/* -p <pagesize> */
char	  ProcessName[50];	/* -P process name, e.g., when using -N. */
int	  RmDir = 0;		/* -R rm -r the home directory. */
int	  Quiet = 0;		/* -q Limit the number of messages. */
int	  NReaders = 0;		/* -r #threads reading records. */
int	  NSlices = -1;		/* -s <#slices> */
int	  Seed;			/* -S <seed>, defaults varies by pid & time. */
int	  ThreadFlag = 0;	/* -T Share handles with DB_THREAD. PARTIAL */
int	  Trickle = 0;		/* -t Run a memp_trickle() thread. */
int	  Update = 0;		/* -u writer threads just replace iecords. */
int	  Verbose = 0;		/* -v verbose messages. */
int	  Verify = 0;		/* -V At exit, do on-disk records == in-mem? */
int	  NWriters = 4;		/* -w #threads inserting or deleting records.*/
int	  Yield = 0;		/* -y Increase collisions and transaction aborts
				   by yielding the cpu after each api call. */
int	  SubProcess = 0;	/* -Z */

/* Per-thread statistics */
struct _statistics {		/* These are set by reader and/or writers. */
	int	aborts;		/* Read/write. */
	int	adds;		/* Write. */
	int	updates;	/* Write. */
	int	deletes;	/* Write. */
	int	txns;		/* Write. */
	int	found;		/* Read. */
	int	notfound;	/* Read. */
} *Perf;

const char *Progname = "ex_slice";

#define	DATABASE	"slice.db"		/* Database name. */
#define	WORDLIST	"../test/tcl/wordlist"	/* Dictionary. */


DB        *Dbp;			/* Database handle. */
DB_ENV    *DbEnv;		/* Database environment. */
DB_ENV   **Slices;		/* Handles of the slices, for statistics. */
int	   ActiveThreads;	/* The current number of worker threads. */
time_t	   EndTime;		/* Set the end time of the run for -d <dur>. */
int	   KeepFsync = 0;	/* Skip fsync if durability is < system crash */
int	   NThreads;		/* Total threads. */
int	   Quit = 0;		/* Interrupt handling flag. */
time_t	   Setup;		/* Time at the end of setup, start of work. */
long	   CpuUse;		/* Milliseconds of 'work' from getrusage(). */

/* Per-thread activity statistics for each slice. */
typedef struct slicestats {
	DB_MPOOL_STAT	*mp_stats;
	DB_LOCK_STAT	*lock_stats;
	DB_TXN_STAT	*txn_stats;
} SliceStats_t;

SliceStats_t *SliceStats = NULL;

struct entry {
	char *word;
	short present;
};

struct entry *WordList;					/* Word list. */

/*
 * get_time_ms
 *	Return the number of milliseconds since the first call by the process.
 *
 *	This wraps and mismeasures intervals longer than ~~ 24 days.
 */
long
get_time_ms()
{
	static struct timeval start;
	struct timeval now;
	long millis;

	if (start.tv_sec == 0) {
		(void)gettimeofday(&start, NULL);
		return (0);
	}
	(void)gettimeofday(&now, NULL);
	/* Adjust for e.g., start = 1.99, now = 2.01 */
	if (now.tv_usec < start.tv_usec) {
		now.tv_usec += 1000000;
		now.tv_sec -= 1;
		assert(now.tv_usec > start.tv_usec);
	}
	millis = (now.tv_sec - start.tv_sec) * 1000 +
		(now.tv_usec - start.tv_usec) / 1000;
	return (millis);
}


/*
 * get_cpu_ms
 *	Return the number of milliseconds of cpu used by this process.
 *
 *	This wraps around and mismeasures cpu usage of more than ~~ 24 days.
 */
long
get_cpu_ms()
{
	struct rusage now;
	long millis;

	(void)getrusage(RUSAGE_SELF, &now);
	millis = (now.ru_utime.tv_sec + now.ru_stime.tv_sec) * 1000 +
		 (now.ru_utime.tv_usec + now.ru_stime.tv_usec) / 1000;
	return (millis);
}

/*
 * get_scaled --
 *	Added scaling suffix [kmg] support to atoll() --
 *	These are multiples of 1000, not 1024.
 *
 *	This returns -1 on error, which is not a valid value
 *	for options that call this function.
 */
long long
get_scaled(char *s)
{
	long long scale, val;
	char *endp;

	val = strtoll(s, &endp, 0);
	switch (*endp) {
	case 'k':
	case 'K':
		scale = 1000;
		break;
	case 'm':
	case 'M':
		scale = 1000000;
		break;
	case 'g':
	case 'G':
		scale = 1000000000;
		break;
	default:
		scale = 1;
		break;
	}
	if (val > INT64_MAX / scale) {
		fprintf(stderr,
		    "%s would overflow a signed 64 bit integer\n",s);
		return (-1);
	}
	return (val * scale);
}


/*
 * print_help --
 *	Print a help message about all options, sorted by the option name.
 */
void
print_help()
{
	printf("ex_slice \\\n");
	printf("  -C\t\tPerform a checkpoint at the end of the run.\n");
	printf("  -c <str>\tAdd <str> to to DB_CONFIG; may be repeated.\n");
	printf("  -D [nsw]\tChange transaction durability to one of DB_TXN_{NOSYNC,SYNC,WRITE_NOSYNC} [default: write_nosync].\n");
	printf("  -d <#secs>\tSet the duration of this run, in seconds [default: 10].\n");
	printf("  -H\t\tCreate DB_HASH files [default: DB_BTREE].\n");
	printf("  -h <home>\tSet the environment home directory [default: SLICE].\n");
	printf("  -k [smw]\tSet the slice key callback function to single or multi-key [default: whole key].\n");
	printf("  -L\t\tTurn on the thread tracking and the locker cache.\n");
	printf("  -M <#>\tRun in multiple processes, splitting up work [default: just one process].\n");
	printf("  -N\t\tDo not create the on-disk environment.\n");
	printf("  -n <#>[kmg]\tSet the number of distinct record values [default: 1M].\n");
	printf("  -O <extra>\tSet the overflow record size increase [default: 0].\n");
	printf("  -o <#>[kmg]\tDo this many write transactions [default: 1M].\n");
	printf("  -P <title>\tSet the name of this process in its messages.\n");
	printf("  -p <pgsize>\tSet the database pagesize [default: 8K].\n");
	printf("  -q\t\tQuiet mode - reduce the amount of informational messages.\n");
	printf("  -R\t\trm -r the home directory before starting.\n");
	printf("  -r <#readers>\tSet the number of threads which just read single records [default: 0].\n");
	printf("  -S <seed>\tSet the random number seed [default: getpid() ^ time()].\n");
	printf("  -s <#slices>\tSets the number of slices [default: 2 * writer threads)].\n");
	printf("  -T\t\tShare a DB_THREAD'd environment handle [each thread has its own].\n");
	printf("  -u\t\tUpdate records instead of inserting and deleting them.\n");
	printf("  -V\t\tVerify that db results records match in-memory data.\n");
	printf("  -v\t\tSet verbose messages; repeat for additional details.\n");
	printf("  -w <#writers>\tSet the number of threads which write to the db.\n");
	printf("  -y\t\tYield the cpu frequently to encourage contention.\n");
	printf("\n");
}

#define PROPT_BOOL(optchar, var )					\
	if (var) 							\
		printf("-%c ", optchar);				\

#define PROPT_INT(optchar, var )					\
	if (var) 							\
		printf("-%c %d ", optchar, var);			\

#define PROPT_LL(optchar, var )						\
	if (var) 							\
		printf("-%c %lld ", optchar, var);			\

#define PROPT_STR(optchar, var )					\
	if (var != NULL) 						\
		printf("-%c \"%s\" ", optchar, var);			\

/*
 * print_options --
 *	Print the currently set options that do not have the default value.
 */
void
print_options()
{
	char	*begin, *end;

	PROPT_BOOL('C', Checkpoint);

	/* ConfigLines is complex, print each 'line' as a separate -c option. */
	if ((begin = ConfigLines) != NULL) {
		while (*begin != '\0') {
			if ((end = strchr(begin, '\n')) == NULL) {
				printf("-c \"%s\" ", begin);
			    break;
		    } else {
				printf("-c \"%.*s\" ",
				    (int)(end - begin), begin);
				begin = end + 1;
			}
		}
	}

	if (LogWriteMode != NULL)
		PROPT_STR('D', LogWriteMode);
	PROPT_INT('d', Duration);
	if (DbType != DB_BTREE)
    		printf("-H ");
	PROPT_STR('h', Home);
	if (KeyType != SliceKeyWhole)
		printf("-k %c ", KeyType == SliceKeySingle ? 's' : 'm');
	PROPT_BOOL('L', LockerCache);
	PROPT_INT('M', MultiProc);
	PROPT_BOOL('N', NoCreate);
	PROPT_LL('n', NRecords);
	PROPT_INT('O', OverflowSize);
	PROPT_LL('n', OpCount);
	if (ProcessName[0] != '\0')
		printf("-P%s ", ProcessName);
	PROPT_INT('p', Pagesize);
	PROPT_BOOL('q', Quiet);
	PROPT_BOOL('R', RmDir);
	PROPT_INT('r', NReaders);
	PROPT_INT('S', Seed);
	PROPT_INT('s', NSlices);
	PROPT_BOOL('T', ThreadFlag);
	PROPT_BOOL('u', Update);
	PROPT_INT('w', NWriters);
	PROPT_BOOL('y', Yield);
	printf("\n");
}

/*
 * nop -
 *	Do nothing, successfully, for db_env_set_func_xxxxx() calls;
 *	in particular, for experimenting with turning Berkeley DB's calls
 *	to fsync() into no-ops.
 */
int nop()
{
	return (0);
}

/*
 * start_processes -
 *	Create new argv's for the subprocesses created by -M, then exec and
 *	wait for them.
 */
int
start_processes(int argc, char **argv)
{
	int *pids;
	int i, procs_left, status;
	char **nextargv, *newargv[MAXARGC + 10];
	char name_opt[20], opcount_opt[20], r_opt[20], s_opt[20], w_opt[20];

	if (argc > MAXARGC)
		fatal("-M <#procs> command lines are limited to %d options",
		    0, MAXARGC);
	if ((NReaders != 0 && MultiProc > NReaders) || 
	    (NWriters != 0 && MultiProc > NWriters) ||
	    (OpCount != 0 && MultiProc > OpCount))
		fatal("-M%d -r%d -w%d would leave processes without any tasks",
		    0, MultiProc, NReaders, NWriters);

	pids = calloc(MultiProc + 1, sizeof(int));

	nextargv = &newargv[0];
	*nextargv++ = argv[0];
	*nextargv++ = "-ZN";
	memmove(nextargv, &argv[1], (argc - 1) * sizeof(char*));
	nextargv += argc - 1;
	*nextargv++ = r_opt;
	*nextargv++ = w_opt;
	*nextargv++ = s_opt;
	*nextargv++ = name_opt;
	*nextargv++ = opcount_opt;
	*nextargv++ = NULL;
	snprintf(s_opt, sizeof(s_opt), "-s%d", NSlices);
	assert(nextargv < &newargv[sizeof(newargv) / sizeof(char **)]);
	for (i = 1, procs_left = MultiProc;
	    procs_left > 0;
	    i++, procs_left--) {
		/*
		 * This distributes the reader and writer threads fairly
		 * evenly. Any 'partial' thread counts are assigned to
		 * the later processes. E.g., -w4 with 3 processes:
		 * i	 procs_left  NWriters  #write threads
		 * 1		3	4	 4/3 -> -w1
		 * 2		2	3	 3/2 -> -w1
		 * 3		1	2	 2/1 -> -w2
		 */
		snprintf(r_opt, sizeof(r_opt),
		    "-r%d", NReaders / procs_left);
		snprintf(w_opt, sizeof(w_opt),
		    "-w%d", NWriters / procs_left);
		snprintf(opcount_opt, sizeof(opcount_opt),
		    "-o%lld", OpCount / procs_left);
		NReaders -= NReaders / procs_left;
		NWriters -= NWriters / procs_left;
		OpCount -= OpCount / procs_left;
		snprintf(name_opt, sizeof(name_opt),
		    "-P%sproc#%d", ProcessName, i);
		if ((pids[i] = vfork()) == 0) {
			if (execv(argv[0], newargv) < 0)
				fatal("Could not exec subprocess #%d",
				    errno, i);
		} else if (pids[i] == -1)
			fatal("Could not vfork subprocess #%d",
			    errno, i);
	}
	for (i = 1; i <= MultiProc; i++) {

		while (waitpid(pids[i], &status, 0) == -1)
			continue;
		if (status != 0)
			printf("Process #%d pid %d returned %d\n", i, pids[i], status);

	}
	printf("All child processes exited.\n");
	return (0);
}

/*
 * ex_slice --
 *	Run a simple threaded application of some numbers of readers and
 *	writers competing for a set of words.
 */
int
main(argc, argv)
	int argc;
	char *argv[];
{
	extern char *optarg;
	extern int errno, optind;
	FILE *fp;
	pthread_t *tids;
	time_t begin, work_done;
	int ch, i, ret;
	unsigned long long data_total, cache_size;
	char buf[200], *s;
	void *retp;

	setlinebuf(stdout);
	Seed = getpid() ^ time(NULL);
	begin = get_time_ms();
	/* The first call starts the timer. */
	(void)get_time_ms();
	while ((ch = getopt(argc,
	    argv, "Cc:D:d:Hh:k:LM:Nn:O:o:P:p:qRr:S:s:TtuVvw:yZ")) != EOF)
		switch (ch) {
		case 'C':
			Checkpoint = 1;
			break;
		case 'c':
			/* -c <str> Append <str> to any previous ones. */
			if (ConfigLines == NULL) {
				if ((ConfigLines = strdup(optarg)) == NULL)
					fatal("Cannot strdup %s",
					    errno, optarg);
			} else {
				i = strlen(ConfigLines);
				if ((ConfigLines = realloc(ConfigLines,
				    i + 1 + strlen(optarg))) == NULL)
					fatal("Cannot malloc %d ", errno, i);
				ConfigLines[i] = '\n';
				strcpy(ConfigLines + i + 1, optarg);
			}
			break;
		case 'D':	/* -D {sync, no_sync, write_nosync} */
			/* Look at just the first character. */
			switch (*optarg) {
			  case 'n':
			  case 'N':
			  	LogWriteMode = "nosync";
				break;
			  case 's':
			  case 'S':
			  	LogWriteMode = NULL;
				KeepFsync = 1;
				break;
			  case 'w':
			  case 'W':
			  	LogWriteMode = "write_nosync";
				break;
			  default:
			  	return (usage(ch, optarg));
			}
			break;
		case 'd':
			if ((Duration = atoi(optarg)) <= 0)
				return (usage(ch, optarg));
			break;
		case 'H':
			DbType = DB_HASH;
			break;
		case 'h':
			Home = optarg;
			break;
		case 'k':
			/*
			 * Override the whole-key default slice callback
			 *	s - single field ending at '-'
			 *	m - two fields, the above + the third '-'
			 *	w - the whole key
			 */
			switch (*optarg) {
			case 's':
				KeyType = SliceKeySingle;
				break;
			case 'm':
				KeyType = SliceKeyMultiple;
				break;
			case 'w':
				KeyType = SliceKeyWhole;
				break;
			default:
				return (usage(ch, optarg));
			}
			break;
		case 'L':	/* Call set_thread_count() to cache lockers. */
			LockerCache = 1;
			break;
		case 'M':
			if (SubProcess)
				break;
			if ((MultiProc = atoi(optarg)) <= 0 || MultiProc > 1024)
				return (usage(ch, optarg));
			break;
		case 'N':
			NoCreate = 1;
			break;
		case 'n':
			if ((NRecords = get_scaled(optarg)) <= 0)
				return (usage(ch, optarg));
			if (NRecords > 100*1000000)
				printf(
"Caution: generating %lld records may causing excessive paging\n", NRecords);
			break;
		case 'O':
			if ((OverflowSize = atoi(optarg)) <= 0)
				return (usage(ch, optarg));
			if (NRecords > 100*1000000)
				printf(
"Caution: generating %lld records may causing excessive paging\n", NRecords);
			break;
		case 'o':
			/*
			 * -o 0 means to run until interrupted or the
			 * -d duration (EndTime) is reached.
			 */
			if ((OpCount = get_scaled(optarg)) < 0)
				return (usage(ch, optarg));
			break;
		case 'P':
			snprintf(ProcessName, sizeof(ProcessName), "%s: ", optarg);
			break;
		case 'p':
			if ((Pagesize = atoi(optarg)) < 512 || Pagesize > 65536)
				return (usage(ch, optarg));
			break;
		case 'q':
			Quiet = 1;
			break;
		case 'R':
			RmDir = 1;
			break;
		case 'r':
			NReaders = atoi(optarg);
			break;
		case 'S':
			Seed = atoi(optarg);
			break;
		case 's':
			if ((NSlices = atoi(optarg)) < 0)
				return (usage(ch, optarg));
			break;
		case 'T':
			ThreadFlag = DB_THREAD;
			break;
		case 't':
			if (!SubProcess)
				Trickle = 1;
			break;
		case 'u':
			Update = 1;
			break;
		case 'V':
			/*
			 * Comparing the in-memory to database data is reliable
			 * only when we create and fill the database.
			 */
			if (!NoCreate)
				Verify = 1;
			break;
		case 'v':
			/* -v for txn info, 2X messages, 3X for DB_VERB_SLICE */
			Verbose++;
			break;
		case 'w':
			if ((NWriters = atoi(optarg)) < 0)
				return (usage(ch, optarg));
			break;
		case 'y':
			Yield = 1;
			break;
		case 'Z':
			/* Private option: parent process was given -M<#> */
			SubProcess = 1;
			break;
		case '?':
		default:
			print_help();
			return (EXIT_FAILURE);
		}

	if (argc != optind) {
		print_help();
		fatal("Unrecognized extra argument %s", 0, argv[optind]);
	}

	/* By default, run for 10 seconds (or until interrupted). */
	if (OpCount == 0 && Duration == 0)
		Duration = 10;

	/*
	 * Supress the rmdir after seeing a -Z option: we are a -M# sub-process.
	 */
	if (SubProcess)
		RmDir = 0;
	if (NoCreate && Verify)
		fatal("The verify option is safe only when creating the data",
		    0);
	if (NoCreate && RmDir)
		fatal("The remove-directory (-R) option is incompatible with "
		    "do-not-create (-N)", 0);
	if (NReaders + NWriters == 0)
		fatal("There needs to be at least one thread", 0);

	/*
	 * If no -s option has specified a number of slices, default it to
	 * twice the number of writer threads, or 16 if all are readers.
	 */
	if (NSlices < 0 && (NSlices = 2 * NWriters) == 0)
		NSlices = 16;

	/* Initialize the random number generator. */
	srandom(Seed);

	if (SubProcess) {
		printf("%spid %d ", ProcessName, getpid());
		print_options();
	} else {
		printf("ex_slice ");
		print_options();
	}

	if (NSlices != 0 &&
	    (SliceStats = calloc(NSlices, sizeof(*SliceStats))) == NULL)
	    	fatal("Cannot allocate stats for %d slices", errno, NSlices);

	if (RmDir) {
		snprintf(buf, sizeof(buf),
		    "test -d \"%1$s\" && rm -r \"%1$s\"", Home);
		if (system(buf) < 0)
			fatal("Could not remove old home directory: %s", errno);
	}

	/*
	 * Unless LogWriteMode was set to db_txn_sync, change all BDB-initiated 
	 * fsync calls into no-ops. It helps reduce the elapsed time variation 
	 * in cases of less-than-system-crash levels of durability.
	 */
	if (!KeepFsync)
		db_env_set_func_fsync((int (*) __P((int)))nop);

	if (!SubProcess)
	{
	    /*
	     * If <home>/DB_CONFIG does not exist, then create a simple one with
	     * the requested number of slices, each configured with suitable
	     * resources to run the default workload in the Berkeley DB cache.
	     * Also Add any -c <config line> lines>
	     */
	    i = strlen(Home) + 20;
	    if ((s = malloc(i)) == NULL)
		    fatal("malloc(DB_CONFIG path)", errno);
	    snprintf(s, i, "%s/DB_CONFIG", Home);
	    if (access(s, R_OK) < 0) {
		    if (access(Home, R_OK | W_OK | X_OK) < 0 &&
			mkdir(Home, 0777) < 0)
			    fatal("cannot create home directory %s", errno, s);

		    if ((fp = fopen(s, "w")) == NULL) 
			    fatal("cannot create %s", errno, s);
		    /*
		     * The containing environment and all slices share these
		     * logging paramters : 20MB log buffer, 100MB log file size.
		     */
		    fprintf(fp, "set_lg_bsize 20971520\n");
		    fprintf(fp, "set_lg_max 104857600\n");
		    if (LogWriteMode != NULL)
			    fprintf(fp, "set_flags db_txn_%s on\n", LogWriteMode);
		    if (LockerCache)
			    fprintf(fp, "set_thread_count %d\n",
				5 * (NReaders + NWriters) * (1 + MultiProc));

		    /*
		     * Give each item 200 bytes of cache. Allow for each slice
		     * to support an equal part of that, plus 100K base overhead.
		     * This does not account for any OverflowSize; when -O <size>
		     * is set writer() occasionally generates huge records, and
		     * may cause some cache misses.
		     */
		    data_total = NRecords * 200;
		    if (NSlices == 0) {
			    cache_size = data_total + 100 * 1024;
			    fprintf(fp, "set_cachesize %lld %lld 1\n",
				cache_size / GIGABYTE, cache_size % GIGABYTE);
		    } else {
			    /*
			     * Give the 'containing' environment 10MB for cache;
			     * it contains little data. Slices usually get more.
			     */
			    fprintf(fp, "set_cachesize 0 10485760 1\n");
			    fprintf(fp, "set_slice_count %d\n", NSlices);

			    cache_size = data_total / NSlices + 100 * 1024;
			    fprintf(fp, "slice all set_cachesize %lld %lld 1\n",
				cache_size / GIGABYTE, cache_size % GIGABYTE);
		    }
		    if (ConfigLines != NULL)
			    fprintf(fp, "%s\n", ConfigLines);
		    fclose(fp);
	    }
	    free(s);
	}

	/* Configure and open the DB environment, perhaps creating it. */
	if ((ret =
	    db_init(Home, &DbEnv, NoCreate ? 0 : DB_CREATE | DB_RECOVER)) != 0)
		return (EXIT_FAILURE);

	/* Configure and open the database. */
	if ((ret = open_database(DbEnv,
	    DATABASE, &Dbp, NoCreate ? 0 : DB_CREATE)) != 0)
		return (EXIT_FAILURE);
	
	/*
	 * Use -L to fill up the db before starting the timed section of the
	 * program.
	if (!SubProcess && Preload)
		preload(Dbp, 4);
	 */

	/*
	 * Zero out the per-slice statistics measured by this program,
	 * unless the whole environment tree was just created.
	 */
	if (!RmDir && NSlices > 0 && !SubProcess) {
		for (i = 0; i != NSlices; i++) {
			if ((ret = Slices[i]->memp_stat(Slices[i],
			    &SliceStats[i].mp_stats, NULL, DB_STAT_CLEAR)) != 0)
				fatal("memp_stat slice %d failed?", ret, i);
			if ((ret = Slices[i]->lock_stat(Slices[i],
			    &SliceStats[i].lock_stats, DB_STAT_CLEAR)) != 0)
				fatal("lock_stat slice %d failed?", ret, i);
			if ((ret = Slices[i]->txn_stat(Slices[i],
			    &SliceStats[i].txn_stats, DB_STAT_CLEAR)) != 0)
				fatal("lock_stat slice %d failed?", ret, i);
			free(SliceStats[i].lock_stats);
			free(SliceStats[i].mp_stats);
			free(SliceStats[i].txn_stats);
		}
	}

	/* Tell all threads to stop on interrupt, then we'll display statistics. */
	(void)signal(SIGINT, onint);

	/*
	 * With -M# we start up that many processes, distributing the operation
	 * counts and reader and writer threads amongst them by appending -o#
	 * -r#, and -w# options.
	 * The SubProcess and NoCreate (environment) options are always added to
	 * the argument list, as is a -P <process name> option.
	 */
	if (MultiProc != 0)
		start_processes(argc, argv);
	else
	{
		/* Build the key list in memory. */
		word();

		if (Duration)
			EndTime = Duration + time(NULL);
		NThreads = NReaders + NWriters + 2;

		/* Create statistics structures, offset by 1. */
		if ((Perf = calloc(NThreads, sizeof(*Perf))) == NULL)
			fatal("calloc statistics", errno);

		/* Create thread ID structures. */
		if ((tids = malloc(NThreads * sizeof(pthread_t))) == NULL)
			fatal("malloc threads", errno);

		/* Remember when the actual work started. */
		Setup = get_time_ms();
		CpuUse = -get_cpu_ms();

		/* Start the threads which do the actual work. */
		for (i = 0; i < NWriters + NReaders; ++i)
			if ((ret = pthread_create(
			    &tids[i], NULL, tstart, (void *)(uintptr_t)i)) != 0)
				fatal("pthread_create", ret > 0 ? ret : errno);
		ActiveThreads = i;

		/* Create buffer pool trickle thread. */
		if (pthread_create(&tids[i], NULL, trickle, &i))
			fatal("pthread_create", errno);
		++i;

		/*
		 * Create deadlock detector thread. It must keep running until
		 * all threads have finished their update operations, otherwise
		 * some of them could hang in an undetected deadlock.
		 */
		if (pthread_create(&tids[i], NULL, deadlock, &i))
			fatal("pthread_create", errno);

		/*
		 * First wait for the worker threads at the start of the array,
		 * make sure Quit is set and wait for the rest of the threads.
		 */
		for (i = 0; i < NWriters + NReaders; ++i)
			(void)pthread_join(tids[i], &retp);
		/*
		 * By resetting this the deadlock detection thread knows that it
		 * is safe to exit; there are not pending deadlocks.
		 */
		ActiveThreads = 0;
		work_done = get_time_ms();
		CpuUse += get_cpu_ms();
		Quit = 1;
		for (; i < NThreads; ++i)
			(void)pthread_join(tids[i], &retp);

		stats(work_done, 1);

		if (Checkpoint)
			DbEnv->txn_checkpoint(DbEnv, 0, 0, 0);

		if (Verify)
			verify();
	}

	/* Look for signs of performance limitations. */
	if (NSlices > 0 && !SubProcess)
		slice_stats();

	(void)Dbp->close(Dbp, DB_NOSYNC);
	(void)DbEnv->close(DbEnv, 0);

	return (EXIT_SUCCESS);
}

/*      
 * single_slice_field, multiple_slice_field -
 *      Construct the slice specification DBT from the key DBT.
 *
 *	The application selects the field or fields of the key it finds useful
 *	by filling out the 'slice' dbt to point directly into a byte range in
 *	the key dbt or allocating a temporary buffer and copying the relevant
 *	bytes into it.
 *
 * Inputs:
 *      DB      - an open DB handle for a sliced database.
 *      key     - the source of the slice specification field or fields.
 *      slice   - A BDB-owned DBT to be filled in. Most of its fields are
 *		  zeroed. Two fields are set to describe a BDB-provided
 *		  temporary buffer:
 *			slice->size: DB_SLICE_BUFLEN (at least 16)
 *			slice->data: the address of a 32-bit word aligned buffer
 *				     of slice->size bytes.
 *		  The callback may use this buffer when there are more than one
 *		  slice specification fields in the key.
 *
 *      If the slice specifier is a one contiguous region in the key, then:
 *      - Set slice->data to the first byte of that region.
 *      - Set slice->size to the number of significant bytes in that region.
 *      - Change no other fields.
 *      else
 *      - Calculate 'total_size': the space required for all the slice
 *	  specification fields to be concatenated togther into a byte array.
 *	- If total_size is larger than slice->size, then
 *      -	Allocate that much memory, overwriting slice->data. Usually
 *		that space is obtained via malloc(). However, if the application
 *		has configured custom a memory allocator via DB_ENV->set_alloc()
 *		or db_env_set_func_malloc(), then that function must be used. 
 *      - 	Set DB_DBT_APPMALLOC in slice->flags. Berkeley DB will
 *              deallocate the memory before returning from the API call, using
 *		the free function corresponding to the allocator just used.
 *	- Set slice->size to total_size.
 *      - Concatenate the various sections of the slice specification fields
 *        together into slice->data.
 *      - Change no other fields.
 *
 *	The value of slice->size may be zero. It may exceed key->size, if
 *	that is appropriate for the application.
 *
 *  Returns:
 *      0 on success.
 *      an operating system error code, e.g., ENOMEM if malloc() (or any
 *	application specified allocator) fails.
 */
int
single_slice_field(const DB *db, const DBT *key, DBT *slice)
{
	char *s;
	/*
	 * This shows how to handle a simple, one-field slice specifier
	 * which is the initial portion of the key, up to the first '-'.
	 */
	slice->data = key->data;
	if ((s = memchr(key->data, '-', key->size)) == NULL)
		slice->size = key->size;
	else
		slice->size = (u_int32_t)(s - (char *)key->data);
	COMPQUIET(db, NULL);
	return (0);
}

/*
 * multiple_slice_field --
 *	Show how to allocate memory in a slice specification callback function.
 *
 *	This shows how to handle a complex slice specifier which is made from
 *	the first and third comma separated fields in the null-delimited key.
 *
 */
int
multiple_slice_field(const DB *db, const DBT *key, DBT *slice)
{
	const char *f1, *f3, *s;
	size_t l1, l3;

	f3 = NULL;
	l1 = l3 = 0;
	f1 = key->data;
	/*
	* Find the position and length of the first field and, if present,
	* the third field.
	*/
	if ((s = strchr(f1, ',')) == NULL)
		/*
		 * There is no second field, the first goes to the end of the
		 * key.
		 */
		l1 = key->size;
	else if ((f3 = strchr(s, ',')) == NULL)
		/* There is a second field, but no third field. f3 stays empty. */
		l1 = s - f1;
	else if ((s = strchr(f3, ',')) == NULL)
		/* The third field goes to the end of the key. */
		l3 = (f1 + key->size) - f3;
	else 
		l3 = s - f3;

	/* Allocate the memory for the combined slice specifier. */
	slice->size = l1 + l3;
	if ((slice->data = malloc(slice->size)) == NULL)
		return (errno);
	/*
	 * This tells the caller to free the memory once it is no longer needed.
	 */
	slice->flags = DB_DBT_APPMALLOC;

	/* Fill in the field(s). */
	memmove(slice->data, f1, l1);
	if (l3 != 0)
		memmove((char *)slice->data + l1, f3, l3);
	COMPQUIET(db, NULL);
	return (0);
}

/*
 * get_slice_number --
 *	Determine the slice number which handles a key.
 *	It returns -1 for a non-sliced environment.
 *
 *	Perhaps this should be in the API?
 */
int 
get_slice_number(dbp, key)
	DB *dbp;
	DBT *key;
{
	DB *slice_dbp;
	DB_ENV *slice_env;
	int ret;
	const char *slice_home;

	if (NSlices == 0)
		return (-1);
	if ((ret = dbp->slice_lookup(dbp, key, &slice_dbp, 0)) != 0)
		fatal("slice lookup by %s of %s", ret, ProcessName, key->data);
	slice_env = slice_dbp->get_env(slice_dbp);
	if ((ret = slice_env->get_home(slice_env, &slice_home)) != 0)
		fatal("get_home by %s for %s", ret, ProcessName, key->data);
	return (atoi(slice_home + strlen(slice_home) - 3));
}

/*
 * reader -
 *	Entry point of a thread which only reads the database.
 */
int
reader(id)
	int id;
{
	DB *dbp;
	DBT key, data;
	DB_ENV *dbenv;
	int n, ret, t_ret;

	if (ThreadFlag == 0) {
		if ((ret = db_init(Home, &dbenv, 0)) != 0)
			return (ret);
		if ((ret = open_database(dbenv, Dbp->fname, &dbp, 0)) != 0)
			return (ret);
	} else {
		dbenv = DbEnv;
		dbp = Dbp;
	}

	/*
	 * DBT's must use local memory or malloc'd memory if the DB handle
	 * is accessed in a threaded fashion.
	 */
	memset(&key, 0, sizeof(DBT));
	memset(&data, 0, sizeof(DBT));
	data.flags = DB_DBT_MALLOC;

	/*
	 * Read-only threads do not require transaction protection, unless
	 * there's a need for repeatable reads.
	 */
	while (!Quit) {
		/* Pick a key at random, and look it up. */
		n = random() % NRecords;
		key.data = WordList[n].word;
		key.size = strlen(key.data);

		if (Verbose)
			printf("%sreader: %d: list entry %d: %s\n",
			    ProcessName, id, n, WordList[n].word);

		ret = dbp->get(dbp, NULL, &key, &data, 0);
		switch (ret) {
		case DB_LOCK_DEADLOCK:		/* Deadlock. */
			++Perf[id].aborts;
			break;
		case 0:				/* Success. */
			++Perf[id].found;
			free(data.data);
			break;
		case DB_NOTFOUND:		/* Not found. */
			++Perf[id].notfound;
			break;
		default:
			fatal("reader %d: dbp->get: %s",
			    ret, 0, id, (char *)key.data);
		}

		/* This let you run in read-only mode for -d <n> seconds. */
		if (NWriters == 0 && Duration != 0 && time(NULL) > EndTime)
			Quit = 1;
	}
	/*
	 * Close our locally opened database and environment handles without
	 * flushing the changes to stable storage.
	 */
	if (dbp != Dbp)
		ret = dbp->close(dbp, DB_NOSYNC);
	if (dbenv != DbEnv && (t_ret = dbenv->close(dbenv, 0)) != 0 && ret == 0)
		ret = t_ret;
	return (ret);
}

/*
 * writer -
 *	Entry point of a thread which modifies the database.
 */
int
writer(id)
	int id;
{
	DB *dbp;
	DBT key, data;
	DB_ENV *dbenv;
	DB_TXN *txn;
	long now, next_stats, start;
	int n, ret, retries, slice, todo, t_ret;
	unsigned txnid;
	char dbuf[100000], prefix[30];

	start = get_time_ms();
	/* Thread 1 prints stats every 10,000 milliseconds. */
	next_stats = start + 10000;

	todo = 0;
	if (OpCount != 0)
		todo = (OpCount < NWriters) ? OpCount : OpCount / NWriters;

	if (ThreadFlag == 0) {
		if ((ret = db_init(Home, &dbenv, 0)) != 0)
			return (ret);
		snprintf(prefix, sizeof(prefix), "%sw%d", ProcessName, id);
		(void)dbenv->set_errpfx(dbenv, prefix);
		(void)dbenv->set_msgpfx(dbenv, prefix);

		if ((ret = open_database(dbenv, Dbp->fname, &dbp, 0)) != 0)
			return (ret);
	} else {
		dbenv = DbEnv;
		dbp = Dbp;
	}

	/*
	 * DBT's must use local memory or malloc'd memory if the DB handle
	 * is accessed in a threaded fashion.
	 */
	memset(&key, 0, sizeof(DBT));
	memset(&data, 0, sizeof(DBT));
	data.data = dbuf;
	data.ulen = sizeof(dbuf);
	data.flags = DB_DBT_USERMEM;

	while (!Quit) {
		/* Pick a random key. */
		n = random() % NRecords;
		key.data = WordList[n].word;
		key.size = strlen(key.data);
		snprintf(data.data, sizeof(dbuf), "%d", n);

		if (Verbose > 1) {
			slice = get_slice_number(dbp, &key);
			printf("%swriter: %d: slice %d list entry %d: %s\n",
			    ProcessName, id, slice, n, WordList[n].word);
		}

		retries = 0;
		/* Abort and retry. */
		if (0) {
retry:			if ((ret = txn->abort(txn)) != 0)
				fatal("DB_TXN->abort", ret);
			retries++;
			++Perf[id].aborts;
			if (Quit)
				break;
		}

		/* Thread #1 prints out the stats every 10 seconds. */
		if (!Quiet && id == 1) {
			now = get_time_ms();
			if (now >= next_stats) {
				stats(now, 0);
				next_stats = now + 10000;
			}
		}

		/* Begin a transaction, for get + (put or del). */
		if ((ret = dbenv->txn_begin(dbenv, NULL, &txn, 0)) != 0)
			fatal("txn_begin", ret);
		txnid = txn->id(txn);

		/*
		 * Look for the key.  If it doesn't exist, add it.  If it does
		 * exist, either update a field or delete it.
		 */
		ret = dbp->get(dbp, txn, &key, &data, 0);
		switch (ret) {
		case DB_LOCK_DEADLOCK:
			goto retry;
		case 0:
			assert(strcmp(key.data, WordList[n].word) == 0);
			if (Update)
				goto update;
			goto delete;
		case DB_NOTFOUND:
			goto add;
		}

		fatal("writer: %d: dbp->get", ret, id);
		/* NOTREACHED */

add:		/* Add the key with random sized data. */
		data.size = 20 + random() % 128;
		/*
		 * If -o <size> was set, one data item in 30 can be larger than
		 * a page, up to the size of the local data dbt buffer.
		 */
		if (OverflowSize != 0 && random() % 30 == 0 &&
		    (data.size += OverflowSize) > sizeof(dbuf))
		    	data.size = sizeof(dbuf);

		ret = dbp->put(dbp, txn, &key, &data, 0);
		switch (ret) {
		case DB_LOCK_DEADLOCK:
			goto retry;
		case 0:
			++Perf[id].adds;
			WordList[n].present = 1;
			goto commit;
		default:
			fatal("writer: %d: dbp->put(%s)",
			    ret, id, (char *)key.data);
		}

update:		/* Append a transaction number to the data portion. */
		snprintf(data.data, data.size, "%d - #%d", n, Perf[id].txns);
		ret = dbp->put(dbp, txn, &key, &data, 0);
		switch (ret) {
		case DB_LOCK_DEADLOCK:
			goto retry;
		case 0:
			++Perf[id].updates;
			WordList[n].present = 1;
			goto commit;
		}
		
		fatal("writer: %d: dbp->put(%.*s) update existing entry %s txn %x",
		    ret, id, key.size, (char *)key.data,
		    WordList[n].word, txn->txn_slices[0]->txnid);
		/* NOTREACHED */

delete:		/* Delete the key. */
		ret = dbp->del(dbp, txn, &key, 0);
		switch (ret) {
		case DB_LOCK_DEADLOCK:
			goto retry;
		case 0:
			++Perf[id].deletes;
			WordList[n].present = 0;
			goto commit;
		}
		
		fatal("writer: %d: dbp->del(%.*s) entry %s txn %x", ret,
		    id, key.size, (char *)key.data,
		    WordList[n].word, txn->txn_slices[0]->txnid);
		/* NOTREACHED */

commit:		/* The transaction finished, commit it. */
		if ((ret = txn->commit(txn, 0)) != 0)
			fatal("DB_TXN->commit", ret);

		/*
		 * Every time the thread completes 100000 transactions, show
		 * our progress.
		 */
		if (++Perf[id].txns % 100000 == 0) {
			printf(
"%swriter: %2d: adds: %4d: deletes: %4d: updates: %4d aborts: %4d: txns: %4d\n",
			    ProcessName, id, Perf[id].adds, Perf[id].deletes,
			     Perf[id].updates, Perf[id].aborts, Perf[id].txns);
		}

		if (todo > 0 && Perf[id].txns >= todo) {
			now = get_time_ms() - start;
			printf("%swriter %d done: %d txns in %ld.%03ld sec\n",
			    ProcessName, id, Perf[id].txns,
			    now / 1000, now % 1000);
			break;
		}

		/* Complain if this transaction had to retry several times. */
		if (Verbose && retries > 2) {
			printf(
"%swriter: %2d: adds: %4d: deletes: %4d: updates: %4d: aborts: %4d: txns: %4d: "
			    "%d retries @ txn %x slice %d\n",
			    ProcessName, id, Perf[id].adds, Perf[id].deletes,
			    Perf[id].updates, Perf[id].aborts, Perf[id].txns,
			    retries, txnid, get_slice_number(dbp, &key));
		}
	}
	/*
	 * Close our locally opened database and environment handles without
	 * flushing the changes to stable storage.
	 */
	if (dbp != Dbp)
		ret = dbp->close(dbp, DB_NOSYNC);
	if (dbenv != DbEnv && (t_ret = dbenv->close(dbenv, 0)) != 0 && ret == 0)
		ret = t_ret;
	return (ret);
}

/*
 * verify --
 *	At the end of a single-process run of the benchmark, verify that each
 *	key/value pair is present if and only if we expect it to exist.
 */
int
verify()
{
	DBT key, data;
	DB_TXN *txn;
	int i, present, ret;

	ret = present = 0;
	txn = NULL;
	/*
	 * DBT's must use local memory or malloc'd memory if the DB handle
	 * is accessed in a threaded fashion.
	 */
	memset(&key, 0, sizeof(DBT));
	memset(&data, 0, sizeof(DBT));
	data.flags = DB_DBT_MALLOC;

	for (i = 0; i != NRecords; i++) {
		key.data = WordList[i].word;
		key.size = strlen(key.data);
		ret = Dbp->get(Dbp, txn, &key, &data, 0);
		if (WordList[i].present)
			present++;
		if ((ret == 0) != WordList[i].present)
			fatal(
		"Presence mismatch for entry %d: %s expected %s but get %s",
			    i, WordList[i].word,
			    WordList[i].present ? "to see it" : "no record",
			    ret == 0 ? "found it" : db_strerror(ret));
				
	}
	printf("%sVerified %d key-value pairs as expected, %d present\n", ProcessName, i, present);
	return (0);
}


/*
 * stats --
 *	Display reader/writer thread statistics.  To display the statistics
 *	for the mpool trickle or deadlock threads, use db_stat(1).
 */
void
stats(time_sofar, final)
	long time_sofar;
	int final;
{
	int i;
	long long rd_txns, wr_txns;
	char header[100];

	rd_txns = wr_txns = 0;
	time_sofar -= Setup;
	/* Avoid divide by zero errors for intervals less than a millisecond. */
	if (time_sofar <= 0)
		time_sofar = 1;

	for (i = 0; i < NReaders + NWriters;)
		if (++i <= NWriters) {
		    printf(
"%swriter: %2d: adds: %4d: deletes: %4d: updates: %4d: aborts: %4d: txns: %4d\n",
			    ProcessName, i, Perf[i].adds, Perf[i].deletes,
			    Perf[i].updates, Perf[i].aborts, Perf[i].txns);
			wr_txns += Perf[i].txns;
		} else {
			printf(
	"%sreader: %2d: found: %5d: not found: %5d: aborts: %4d\n",
			    ProcessName, i, Perf[i].found,
			    Perf[i].notfound, Perf[i].aborts);
			rd_txns += Perf[i].found + Perf[i].notfound + Perf[i].aborts;
		}

	if (final)
		snprintf(header, sizeof(header),
		    "Overall run %ld.%02ld sec cpu %ld.%02ld:",
		    time_sofar / 1000, (time_sofar % 1000) / 10,
		    CpuUse / 1000, (CpuUse % 1000) / 10);
	else
		snprintf(header, sizeof(header), "At %ld.%02ld sec:",
		    time_sofar / 1000, (time_sofar % 1000) / 10);
		
	printf("%s%s %lld reads @ %.0f reads/sec %lld writes @ %.0f txns/sec\n",
	    ProcessName, header, rd_txns, rd_txns / (time_sofar / 1000.0),
	    wr_txns, wr_txns / (time_sofar / 1000.0));
}

/*
 * slice_stats --
 *	Check the environment's slice statistics for performance-limiting curiousities.
 *
 */
void
slice_stats()
{
	SliceStats_t *stats;
	char cachenotes[256], locknotes[256], txnnotes[256];
	int id, ret;
	unsigned long avg_txns, tot_begins, tot_commits;

	/*
	 * Make note of slices whose lock wait times are vary a lot from others,
	 * or had lots of cache misses, or were unusually busy or idle. The
	 * first loop averages; the second prints the outlying slices' values.
	 */
	tot_begins = tot_commits = 0;
	for (id = 0, stats = SliceStats; id != NSlices; id++, stats++) {
		if ((ret = Slices[id]->memp_stat(Slices[id],
		    &stats->mp_stats, NULL, 0)) != 0)
			fatal("memp_stat slice %d failed", ret, id);
		if ((ret = Slices[id]->lock_stat(Slices[id],
		    &stats->lock_stats, 0)) != 0)
			fatal("lock_stat slice %d failed", ret, id);
		if ((ret = Slices[id]->txn_stat(Slices[id],
		    &stats->txn_stats, 0)) != 0)
			fatal("txn_stat slice %d failed", ret, id);
		tot_commits += SliceStats[id].txn_stats->st_ncommits;
		tot_begins += SliceStats[id].txn_stats->st_nbegins;
	}
	avg_txns = tot_commits / NSlices;
	printf("%sStatistics for %d slices: transaction count %lu\n",
	    ProcessName, NSlices, avg_txns);
	for (id = 0, stats = SliceStats; id != NSlices; id++, stats++) {
		cachenotes[0] = locknotes[0] = txnnotes[0] = '\0';

		/* Any amount of cache misses is noted; the data is too big. */
		if (stats->mp_stats->st_ro_evict >
		    stats->mp_stats->st_pages / 10 || 
		    stats->mp_stats->st_rw_evict >
		    stats->mp_stats->st_pages / 10)
			snprintf(cachenotes, sizeof(cachenotes),
			    "Cache too small: pages %llu page in %llu "
			    "page out %llu evictions: clean %llu dirty %llu; ",
			    (unsigned long long)stats->mp_stats->st_pages,
			    (unsigned long long)stats->mp_stats->st_page_in,
			    (unsigned long long)stats->mp_stats->st_page_out,
			    (unsigned long long)stats->mp_stats->st_ro_evict,
			    (unsigned long long)stats->mp_stats->st_rw_evict);

		/* Call out txn counts more than ~10% away from the average. */
		if (stats->txn_stats->st_nbegins < avg_txns * 0.9 ||
		    stats->txn_stats->st_nbegins > avg_txns * 1.1)
			snprintf(txnnotes, sizeof(txnnotes),
			    "%s transaction count %lu; ",
			    stats->txn_stats->st_nbegins > avg_txns ?  "high" :
			    "low", stats->txn_stats->st_nbegins);

		if (!Quiet && (cachenotes[0] != '\0' ||
		    locknotes[0] != '\0' || txnnotes[0] != '\0'))
			    printf("%sslice %2d: %s%s%s\n", ProcessName, id,
				cachenotes, locknotes, txnnotes);
		free(SliceStats[id].mp_stats);
		free(SliceStats[id].lock_stats);
		free(SliceStats[id].txn_stats);
	}
}

/*
 * db_init --
 *	Open the environment, possibly creating it.
 */
int
db_init(home, dbenvp, open_flags)
	const char *home;
	DB_ENV **dbenvp;
	u_int32_t open_flags;
{
	DB_ENV *dbenv, **slices;;
	int i, len, ret;
	u_int32_t count;
	char *format, *prefix;

	*dbenvp = NULL;
	if ((ret = db_env_create(&dbenv, 0)) != 0) {
		fprintf(stderr,
		    "%s: db_env_create: %s\n", Progname, db_strerror(ret));
		return (EXIT_FAILURE);
	}
	if (Yield)
		(void)dbenv->set_flags(dbenv, DB_YIELDCPU, 1);

	dbenv->set_errfile(dbenv, stderr);
	dbenv->set_errpfx(dbenv, Progname);
	if ((ret = dbenv->set_mp_pagesize(dbenv, Pagesize)) != 0)
		fatal("set_mp_pagesize(%d)", ret, Pagesize);

	/*
	 * Override the default number of mutexes to be used by the cache --
	 * it can be excessive for large configurations.
	 */
	i = NWriters > 10 ? NWriters : 10;
	if ((ret = dbenv->set_mp_mtxcount(dbenv, i)) != 0)
		fatal("set_mp_mtxcount(%d)", ret, i);
	/*(void)dbenv->set_lk_detect(dbenv, DB_LOCK_YOUNGEST);*/
	if (Verbose > 2)
		(void)dbenv->set_verbose(dbenv, DB_VERB_SLICE, 1);

	open_flags |= DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | DB_INIT_TXN;
	open_flags |= ThreadFlag;

	if ((ret = dbenv->open(dbenv, home, open_flags, 0775)) != 0) {
		dbenv->err(dbenv, ret, NULL);
		(void)dbenv->close(dbenv, 0);
		Quit = 1;
		return (EXIT_FAILURE);
	}
	/* The memory for the error prefixes is not freed. */
	i = 20 + strlen(ProcessName);
	prefix = malloc(i);
	snprintf(prefix, i, "%sContainer", ProcessName);
	dbenv->set_errpfx(dbenv, prefix);
	
	/* Give each slice a unique error message prefix. */
	if (dbenv->get_slice_count(dbenv, &count) == 0 && count != 0) {
		if (count != (unsigned)NSlices)
			fatal("db_init: -s option said %d slices, \"%s\" has %d",
			    0, NSlices, home, count);
		format = "[slice %d] ";
		len = strlen(format) + 3;	/* Allow for 999 slices. */
		if ((ret = dbenv->get_slices(dbenv, &slices)) != 0)
			fatal("get_slices failed?", ret);
		if (Slices == NULL)
			Slices = slices;
		for (i = 0; slices[i] != NULL; i++) {
			if ((prefix = malloc(len)) == NULL)
				fatal("malloc prefix of %d bytes", errno, len);
			snprintf(prefix, len, format, i);
			slices[i]->set_errpfx(slices[i], prefix);
		}
	}

	*dbenvp = dbenv;
	return (0);
}

int
open_database(dbenv, name, dbpp, open_flags)
	DB_ENV *dbenv;
	const char *name;
	DB **dbpp;
	u_int32_t open_flags;
{
	DB *dbp;
	DB_TXN *txn;
	int ret;
	u_int32_t slice_count;

	dbp = NULL;
	txn = NULL;
	
	if ((ret = db_create(&dbp, dbenv, 0)) != 0) {
		dbenv->err(dbenv, ret, "db_create");
		(void)dbenv->close(dbenv, 0);
		return (EXIT_FAILURE);
	}
	if ((ret = dbp->set_pagesize(dbp, Pagesize)) != 0) {
		dbp->err(dbp, ret, "set_pagesize(%d)", Pagesize);
		goto err;
	}

	/*
	 * If the environment is sliced then -Ks or -Km can override the
	 * default, whole-key slice callback function.
	 */
	if ((ret = dbenv->get_slice_count(dbenv, &slice_count)) == 0 &&
	    slice_count > 0) {
		if (KeyType != SliceKeyWhole && (ret =
		    dbp->set_slice_callback(dbp, KeyType == SliceKeySingle ?
		    single_slice_field : multiple_slice_field)) != 0) {
			dbp->err(dbp, ret, "set_slice_callback");
			goto err;
		}
		open_flags |= DB_SLICED;
    	}

	/*
	 * Open with a transaction so that all later operations with this database
	 * handle are transactionally protected.
	 */
	if ((ret = dbenv->txn_begin(dbenv, NULL, &txn, 0)) != 0)
		fatal("open_database txn_begin", ret);

	if ((ret =
	    dbp->open(dbp, txn, name, NULL, DbType, open_flags, 0664)) != 0) {
		dbp->err(dbp, ret, "%s: open", name);
		goto err;
	}
	if (txn != NULL && (ret = txn->commit(txn, 0)) != 0)
		fatal("open_database DB_TXN->commit", ret);

	*dbpp = dbp;
	return (0);

err:
	if (txn != NULL)
		txn->commit(txn, 0);
	if (dbp != NULL)
		dbp->close(dbp, 0);
	return (ret);
}

/*
 * tstart --
 *	Thread start function for readers and writers.
 */
void *
tstart(arg)
	void *arg;
{
	pthread_t tid;
	u_int id;

	id = (uintptr_t)arg + 1;

	tid = pthread_self();

	if (id <= (u_int)NWriters) {
		writer(id);
	} else {
		reader(id);
	}

	return (NULL);
}

/*
 * deadlock --
 *	Thread start function for DB_ENV->lock_detect.
 */
void *
deadlock(arg)
	void *arg;
{
	struct timeval delay;
	pthread_t tid;
	long long ms;
	int deadlocks, total;

	COMPQUIET(arg, NULL);
	tid = pthread_self();
	deadlocks = total = 0;
	delay.tv_sec = 0;
	delay.tv_usec = 16000;

	while (ActiveThreads > 0) {
		ms = get_time_ms();
		(void)DbEnv->lock_detect(DbEnv, 0, DB_LOCK_YOUNGEST, &deadlocks);
		total += deadlocks;

		/*
		 * Check every 20ms, or more frequently if more than 1 deadlock
		 * is detected.  The timeval struct needs to be set again for
		 * each call, because some platforms change the timeval to the
		 * amount of unexpired time.
		 */
		if (delay.tv_usec == 0)
			delay.tv_usec = 100;
		if (deadlocks == 0) {
			if (delay.tv_usec < 16000)
				delay.tv_usec *= 2;
		} else {
			if (delay.tv_usec > 2000)
				delay.tv_usec /= 2;
			/*
			printf("%sDetected %d deadlocks in %d ms new delay %d us\n",
			    ProcessName, deadlocks, get_time_ms() - ms, delay.tv_usec);
			*/
		}

		if (select(0, NULL, NULL, NULL, &delay) < 0)
			perror("select");
		if (Duration != 0 && time(NULL) > EndTime)
			Quit = 1;
	}

	return (NULL);
}

/*
 * trickle --
 *	Thread start function for memp_trickle.
 */
void *
trickle(arg)
	void *arg;
{
	pthread_t tid;
	int wrote;
	long now;

	COMPQUIET(arg, NULL);			/* XXX: shut the compiler up. */
	tid = pthread_self();

	if (!Trickle)
		return (NULL);

	while (!Quit) {
		now = get_time_ms();
		(void)DbEnv->memp_trickle(DbEnv, 10, &wrote);
		if (Verbose && wrote != 0)
			printf("%strickle: wrote %d\n", ProcessName, wrote);
		if (wrote == 0) {
			sleep(2);
			sched_yield();
		}
	}

	return (NULL);
}

/*
 * word --
 *	Build the in-memory word list, starting from a dictionary file, then
 *	adding suffixes to those words until reaching the mandated record count.
 */
void
word()
{
	FILE *fp;		/* Read the dictionary with this. */
	long long cnt, nwords;	/* Count number of words. */
	char word[256];		/* Holds the word being read in or created. */
	char *eol;

	/* Look in standard place, or current directory. */
	if ((fp = fopen(WORDLIST, "r")) == NULL &&
	    (fp = fopen("wordlist", "r")) == NULL)
		fatal(WORDLIST, errno);

	if ((WordList = calloc(NRecords, sizeof(WordList[0]))) == NULL)
		fatal("malloc list for %d words", errno, NRecords);

	for (cnt = 0; cnt < NRecords; ++cnt) {
		if (fgets(word, sizeof(word), fp) == NULL)
			break;
		/* Remove trailing newline */
		if ((eol = strrchr(word, '\n')) != NULL)
			*eol = '\0';

		if ((WordList[cnt].word = strdup(word)) == NULL)
			fatal("strdup %s for item #%lld", errno, word, cnt);
	}
	/* Generate additional values until we reach the desired record count. */
	nwords = cnt;
	while (cnt < NRecords) {
		snprintf(word, sizeof(word),
		    "%s-%lld", WordList[cnt % nwords].word, cnt);
		if ((WordList[cnt].word = strdup(word)) == NULL)
			fatal("strdup %s for item #%lld", errno, word, cnt);
		cnt++;
	}
}

/*
 * fatal --
 *	Report a fatal error and quit.
 *	This assembles the text into a single string so that the message is
 *	displayed coherenly even when multiple threads are calling it
 *	at the same time.
 */
void
fatal(const char *msg, int err, ...)
{
	va_list ap;
	char buf[8192];

	snprintf(buf, sizeof(buf),
	    "%sFatal error in %s: ", ProcessName, Progname);

	va_start(ap, err);
	vsnprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), msg, ap);
	va_end(ap);

	if (err != 0)
		snprintf(buf + strlen(buf),
		    sizeof(buf) - strlen(buf), ": %s", db_strerror(err));
	fprintf(stderr, "%s\n", buf);
	exit(EXIT_FAILURE);

	/* NOTREACHED */
}

/*
 * usage --
 *	Usage message.
 */
int
usage(option, bad_value)
	int option;
	const char *bad_value;
{
	fprintf(stderr, "Invalid value for -%c: %s\n", option, bad_value);
	print_help();
	return (EXIT_FAILURE);
}

/*
 * onint --
 *	Interrupt signal handler.
 */
void
onint(signo)
	int signo;
{
	COMPQUIET(signo, 0);
	Quit = 1;
}
